package com.atguigu.flink.timeAndwindow;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * Created by 黄凯 on 2023/6/16 0016 20:29
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 */
public class Flink04_WindowAssigner {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> ds = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                        }
                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

       /* SingleOutputStreamOperator<Event> ds = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );*/


        ds.print("input");

        /**
         * 窗口分配器
         * 非keyBy,所有的数据都会落到同一个窗口中
         * 计数窗口
         */
       /* ds.map(enent -> Tuple2.of("user", 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
//                .countWindowAll(5) //计数滚动
                .countWindowAll(5,3) // 计数滑动
                .sum(1)
                .print();*/

        //时间窗口
        /*ds.map(enent -> Tuple2.of("user", 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .windowAll(
                        TumblingEventTimeWindows.of(Time.seconds(10)) // 10秒事件时间滚动窗口
                )
                .sum(1)
                .print();*/

        // keyBy
        //计数
       /* ds.keyBy(Event::getUser)
//                .countWindow(3)  //计数滚动
                .countWindow(10, 3) //计数滑动
                .sum("ts")
                .print();*/

        //时间
        ds.map(event -> Tuple2.of(event.getUser(), 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t -> t.f0)
                .window(

                        //滚动
//                        TumblingProcessingTimeWindows.of(Time.seconds(10)) //处理时间滚动窗口
//                        TumblingEventTimeWindows.of(Time.seconds(5))  //事件时间滚动窗口

//                        TumblingEventTimeWindows.of(Time.days(1),Time.hours(-8))

                        //滑动
//                        SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)) //处理时间滑动窗口
                        SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5)) //事件时间滑动窗口

                        //会话
//                        ProcessingTimeSessionWindows.withGap(Time.seconds(5)) //处理时间会话窗口
//                        EventTimeSessionWindows.withGap(Time.seconds(5)) //事件时间

                        //全局 了解
//                        GlobalWindows.create()


                ).sum(1)
                .print();

        env.execute();

    }

}
